package net.bw.realtime.dy.common.data;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.*;
import java.time.*;
import java.util.*;

public class TEST {

    private static final String TOPIC = "live_eve";
    private static final String BOOTSTRAP_SERVERS = "node101:9092";
    private static final String[] EVENT_TYPES = {"gift", "message", "enter", "leave", "share", "like"};
    private static final String[] MESSAGES = {
            "精彩! 再来一个 666",
            "主播好厉害s!",
            "关注了，下次还来",
            "礼物走一波~",
            "这个操作太秀了",
            "前排围观"
    };

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            long startTimestamp = LocalDateTime.of(2024, 3, 26, 0, 0, 0)
                    .toInstant(ZoneOffset.UTC).toEpochMilli();

            ObjectMapper mapper = new ObjectMapper();
            Random random = new Random();

            for (int i = 0; i < 10000; i++) {
                Map<String, Object> event = new LinkedHashMap<>();
                // 公共字段
                event.put("eventId", UUID.randomUUID().toString());
                event.put("liveId", "live_001");
                event.put("anchorId", "anchor_1");
                event.put("userId", "user_" + (random.nextInt(100) + 1)); // 100个不同用户
                event.put("roomId", "live_001");
                event.put("ts", startTimestamp);

                // 随机生成事件类型
                String eventType = getRandomEventType(random);
                event.put("eventType", eventType);

                // 特殊字段处理
                switch (eventType) {
                    case "gift":
                        event.put("giftId", random.nextInt(20) + 1);  // 1-20的礼物ID
                        event.put("giftValue", random.nextInt(1000) + 1);
                        break;
                    case "message":
                        event.put("messageContent",
                                MESSAGES[random.nextInt(MESSAGES.length)]);
                        break;
                    // enter/leave/share/like 无额外字段
                }

                // 序列化并发送
                producer.send(new ProducerRecord<>(TOPIC, mapper.writeValueAsString(event)));

                // 时间戳递增（每秒1个事件）
                startTimestamp += 1000;
            }

            producer.flush();
            System.out.println("成功生成10000条测试数据");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static String getRandomEventType(Random random) {
        // 调整事件类型概率（示例比例）
        double prob = random.nextDouble();
        if (prob < 0.3) return "message";   // 30%
        if (prob < 0.6) return "gift";       // 30%
        if (prob < 0.7) return "enter";      // 10%
        if (prob < 0.8) return "leave";     // 10%
        if (prob < 0.9) return "share";     // 10%
        return "like";                      // 10%
    }
}